package cn.careerx.app.common.models;

import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;

import libs.F.Action2;
import libs.chash.HashAlgorithm;
import libs.chash.KetamaNodeLocator;
import libs.chash.Node;
import libs.chash.NodeImpl;

import org.apache.commons.io.IOUtils;

import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

import com.qiniu.util.StringUtils;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;

public class Redis {
	
	private final static Map<String, KetamaNodeLocator<String, JedisPool>> COMP_LOCATORS = new ConcurrentHashMap<>();
	
	private final static ReentrantLock initLock = new ReentrantLock();
	
	private static Config redisConfig;
	
	private static KetamaNodeLocator<String, JedisPool> createJedisPoolLocator(List<?> confs){
		List<Node<String, JedisPool>> nodes = new ArrayList<>();
		for(Object confObj : confs){
			Config conf = (Config)confObj;
			JedisPoolConfig config = new JedisPoolConfig();
	        config.setMaxTotal(128);
	        config.setMaxIdle(4);
	        config.setMaxWaitMillis(20000);
	        config.setTestOnBorrow(true);
	        String poolName = conf.getString("name");
	        JedisPool jedisPool = new JedisPool(config, conf.getString("host"), conf.getInt("port"), 5000, conf.hasPath("password")?conf.getString("password"):null);
	        nodes.add(new NodeImpl<String, JedisPool>(poolName, jedisPool));
		}
		return new KetamaNodeLocator<>(nodes, HashAlgorithm.KETAMA_HASH, 160);
	}
	
	private static Action2<Node<String,JedisPool>, List<JedisPool>> COLLECT_ALL_ACTION = new Action2<Node<String,JedisPool>, List<JedisPool>>() {

		@Override
		public void invoke(Node<String, JedisPool> node, List<JedisPool> list) {
			list.add(node.getObject());
		}
	};
	
	public static List<JedisPool> getJedisPools(String component){
		KetamaNodeLocator<String, JedisPool> locator = COMP_LOCATORS.get(component);
		if(locator == null){
			initLock.lock();
			locator = COMP_LOCATORS.get(component);
			try{
				if(redisConfig == null){
					InputStream is = Redis.class.getClassLoader().getResourceAsStream("redis.conf");
					if(is != null){
						List<String> lines;
						try {
							lines = IOUtils.readLines(is, "utf-8");
							String lineSeparator = System.getProperty("line.separator");
							String confStr = StringUtils.join(lines, lineSeparator);
							redisConfig = ConfigFactory.parseString(confStr);
							if(redisConfig.hasPath(component)){
								locator = createJedisPoolLocator(redisConfig.getConfigList(component));
								COMP_LOCATORS.put(component, locator);
							}
						} catch (IOException e) {}
						finally{
							IOUtils.closeQuietly(is);
						}
					}
				}
			}finally{
				initLock.unlock();
			}
		}
		if(locator == null){
			return null;
		}
		
		List<JedisPool> ret = new ArrayList<>();
		locator.forEachAction2(COLLECT_ALL_ACTION, ret);
		
		return ret;
	}
	
	public static JedisPool getJedisPool(String component, String partitionKey){
		KetamaNodeLocator<String, JedisPool> locator = COMP_LOCATORS.get(component);
		if(locator == null){
			initLock.lock();
			locator = COMP_LOCATORS.get(component);
			try{
				if(redisConfig == null){
					InputStream is = Redis.class.getClassLoader().getResourceAsStream("redis.conf");
					if(is != null){
						List<String> lines;
						try {
							lines = IOUtils.readLines(is, "utf-8");
							String lineSeparator = System.getProperty("line.separator");
							String confStr = StringUtils.join(lines, lineSeparator);
							redisConfig = ConfigFactory.parseString(confStr);
							if(redisConfig.hasPath(component)){
								locator = createJedisPoolLocator(redisConfig.getConfigList(component));
								COMP_LOCATORS.put(component, locator);
							}
						} catch (IOException e) {}
						finally{
							IOUtils.closeQuietly(is);
						}
					}
				}
			}finally{
				initLock.unlock();
			}
		}
		if(locator == null){
			return null;
		}
		Node<String, JedisPool> node = locator.getPrimary(partitionKey);
		if(node == null){
			return null;
		}
		return node.getObject();
	}
	
	public static Map<JedisPool, List<String>> getGroupedJedisPools(String component, Iterable<String> partitionKeys){
		Map<JedisPool, List<String>> ret = new HashMap<>();
		if(partitionKeys == null){
			return ret;
		}
		KetamaNodeLocator<String, JedisPool> locator = COMP_LOCATORS.get(component);
		if(locator == null){
			initLock.lock();
			locator = COMP_LOCATORS.get(component);
			try{
				if(redisConfig == null){
					InputStream is = Redis.class.getClassLoader().getResourceAsStream("redis.conf");
					if(is != null){
						List<String> lines;
						try {
							lines = IOUtils.readLines(is, "utf-8");
							String lineSeparator = System.getProperty("line.separator");
							String confStr = StringUtils.join(lines, lineSeparator);
							redisConfig = ConfigFactory.parseString(confStr);
							if(redisConfig.hasPath(component)){
								locator = createJedisPoolLocator(redisConfig.getConfigList(component));
								COMP_LOCATORS.put(component, locator);
							}
						} catch (IOException e) {}
						finally{
							IOUtils.closeQuietly(is);
						}
					}
				}
			}finally{
				initLock.unlock();
			}
		}
		if(locator == null){
			return ret;
		}
		
		for(String pKey : partitionKeys){
			Node<String, JedisPool> node = locator.getPrimary(pKey);
			if(node == null){
				continue;
			}
			List<String> groupedKeys = ret.get(node.getObject());
			if(groupedKeys == null){
				groupedKeys = new ArrayList<>();
				ret.put(node.getObject(), groupedKeys);
			}
			groupedKeys.add(pKey);
		}
		return ret;
	}
}
